package com.gupaoedu.kafka.kafkapractice;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @Author: JLL
 * @Date: 2019/10/31 0:39
 */
public class GpKafkaConsumer extends  Thread{

    KafkaConsumer<Integer,String> consumer;
    String topic;

    public GpKafkaConsumer(String topic) {
        Properties properties = new Properties();
        //连接字符串
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.150.130:9092");
        //client_id
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG,"gp-producer");

        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"gp-gid");

        //这里是获得消息的 超时时间。肯定是要配的
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");
        //自动提交的间隔。  什么叫自动提交？？？
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");

        //key的 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        //value的 反序列化 、 这些都是告诉producer用什么序列化的方式去传输
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

        consumer = new KafkaConsumer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {
        //消费者 去 订阅消息
        consumer.subscribe(Collections.singleton(this.topic));
        while (true){
            //拉取数据，超时时间是1秒
            ConsumerRecords<Integer,String> consumerRecords =
                    consumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(record -> {
                System.out.println(record.key() + " " +record.value() + " " + record.offset());
            });

        }
    }


    public static void main(String[] args) {
        new GpKafkaConsumer("test").start();
    }
}
